package com.atguigu.gmall.rabbit.config;

import com.alibaba.fastjson.JSON;
import com.atguigu.gmall.rabbit.service.RabbitService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;

import javax.annotation.PostConstruct;
import java.util.concurrent.TimeUnit;

/**
 * 对生产者使用模板对象进行设置生产者回调
 * 1.在配置文件中开启生产者确认
 * publisher-confirm-type: correlated  #设置交换机异步回调
 * publisher-returns: true            #开启队列异常回调
 * 2.在代码中对模板对象RabbitTemplate设置两个回调方法
 * 交换机回调（包含正常，异常情况）
 * 队列异常回调（交换机路由消息失败）
 *
 * @author: atguigu
 * @create: 2023-08-11 14:08
 */
@Slf4j
@Configuration
public class MQProducerConfig implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {


    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private RedisTemplate redisTemplate;


    /**
     * 项目启动后会自动执行该方法
     */
    @PostConstruct
    public void init() {
        log.info("项目启动后，进行了RabbitTemplate模板对象设置回调");
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
    }


    /**
     * 交换机确认：当生产者发送消息正常到达交换机，一级交换机未收到消息，都会执行该回调方法
     *
     * @param correlationData 相关数据 当生产者发送时设置“相关数据”才会有值
     * @param ack             true:消息正常发送交换机  false：消息发送到交换机失败
     * @param cause           当消息发送交换机失败，包含失败信息
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (!ack) {
            //当消息发送交换机失败，进行重试
            log.error("[生产者-交换机异常确认]发送消息到交换机失败:{}", cause);
            //TODO 只要得到相关数据，就可以进行消息重发
            String id = correlationData.getId();
            this.retrySendMsg(id);
        }
    }

    /**
     * 队列确认：只有当交换机路由消息到队列失败，才会走队列确认回调方法
     *
     * @param message    回调消息对象
     * @param replyCode  应答码
     * @param replyText  应答文本.
     * @param exchange   交换机名称.
     * @param routingKey 路由键.
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.error("[生产者-队列异常确认],消息路由queue失败，应答码={}，原因={}，交换机={}，路由键={}，消息={},进行生产者消息重发",
                replyCode, replyText, exchange, routingKey, message.toString());
        //通过message消息对象属性中获取头信息“spring_returned_message_correlation”获取相关数据ID，通过该ID查询Redis得到所有相关数据，就可以进行消息重发
        String id = message.getMessageProperties().getHeader("spring_returned_message_correlation");
        //todo 解决方案一：判断是否为延迟消息，如果是延迟消息不进行重发 1.方式1判断相关数ID是否有值  2.方式2发送延迟消息也设置相关数据-延迟类型
        if (StringUtils.isNotBlank(id)) {
            this.retrySendMsg(id);
        }
    }


    /**
     * 生产者消息重发：将原本发送失败消息再次进行发送
     *
     * @param id
     */
    private void retrySendMsg(String id) {
        //1.构建相关数据存放在Redis中key
        String key = "mq:" + id;
        //2.查询Redis中相关数据
        String gmallCorrelationDataStr = (String) redisTemplate.opsForValue().get(key);
        GmallCorrelationData correlationData = JSON.parseObject(gmallCorrelationDataStr, GmallCorrelationData.class);

        //方案二：方式一：如果是延迟消息 不重发
        if (correlationData.isDelay()) {
            return;
        }

        //3.判断发送次数 最多重发三次
        if (correlationData.getRetryCount() > 2) {
            log.error("[消息重发]超过上限，及时通知运维，或者将异常业务数据进行写库记录异常");
            redisTemplate.delete(key);
            return;
        }
        //4.进行消息再次发送
        //4.1 更新Redis中重试次数，同时记得设置过期时间
        correlationData.setRetryCount(correlationData.getRetryCount() + 1);
        log.info("[消息重发]正在进行消息第{}次重发", correlationData.getRetryCount());
        //获取key有效时间
        Long expire = redisTemplate.getExpire(key);
        redisTemplate.opsForValue().set(key, JSON.toJSONString(correlationData), expire, TimeUnit.SECONDS);
        //4.2 先对原本消息再次重发-记得设置相关数据
        if (correlationData.isDelay()) {
            //4.2.1 如果是延迟消息-重发带相关数据同时也需要延迟时间
            rabbitTemplate.convertAndSend(correlationData.getExchange(), correlationData.getRoutingKey(), correlationData.getMessage(), message -> {
                message.getMessageProperties().setDelay(correlationData.getDelayTime() * 1000);
                return message;
            }, correlationData);
        } else {
            //4.2.2 如果是普通消息-重发带相关数据即可
            rabbitTemplate.convertAndSend(correlationData.getExchange(), correlationData.getRoutingKey(), correlationData.getMessage(), correlationData);
        }
    }
}
